package com.atguigu.gmall.realtime.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.gmall.realtime.bean.{PageActionLog, PageLog, StartLog}
import com.atguigu.gmall.realtime.utils.MyKafkaUtils
import java.lang
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 日志数据消费分流
 * 1.准备实时环境
 * 2.从kafka中消费数据
 * 3.处理数据
 *   3.1 转换数据结构
 *   3.2 数据分流
 */
object OdsBaseLogApp {
    def main(args: Array[String]): Unit = {
        // 1.准备实时环境
        val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_log_app").setMaster("local[4]")
        val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

        // 2.从kafka中消费数据
        val topic = "ODS_BASE_LOG_1216"
        val groupId = "ODS_BASE_LOG_GROUP"
        val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId)

        // 3.处理数据
        // 3.1 转换数据结构
        val jsonObjDStream: DStream[JSONObject] = kafkaDStream.map(
            consumerRecord => {
                // 提取出数据
                val logData: String = consumerRecord.value()
                // 转换数据为JSON格式
                val jsonObj: JSONObject = JSON.parseObject(logData)
                jsonObj
            }
        )
        // jsonObjDStream.print(100)

        // 3.2 分流
        val DWD_ERROR_TOPIC: String = "DWD_ERROR_TOPIC_1216"
        val DWD_PAGE_TOPIC = "DWD_PAGE_TOPIC_1216"
        val DWD_PAGE_ACTION_TOPIC = "DWD_PAGE_ACTION_TOPIC_1216"
        val DWD_START_TOPIC = "DWD_START_TOPIC_1216"

        jsonObjDStream.foreachRDD(
            rdd => {
                rdd.foreach(
                    jsonObj => {
                        val errObj: JSONObject = jsonObj.getJSONObject("err")
                        if (errObj != null) {
                            // 发往错误主题中
                            MyKafkaUtils.send(DWD_ERROR_TOPIC, jsonObj.toJSONString)
                        } else {
                            // 提取公共字段
                            val commonObj: JSONObject = jsonObj.getJSONObject("common")
                            val ar: String = commonObj.getString("ar")
                            val ba: String = commonObj.getString("ba")
                            val ch: String = commonObj.getString("ch")
                            val isNew: String = commonObj.getString("is_new")
                            val md: String = commonObj.getString("md")
                            val mid: String = commonObj.getString("mid")
                            val os: String = commonObj.getString("os")
                            val uid: String = commonObj.getString("uid")
                            val vc: String = commonObj.getString("vc")

                            // 提取ts
                            val ts: lang.Long = jsonObj.getLong("ts")

                            val pageObj: JSONObject = jsonObj.getJSONObject("page")
                            if (pageObj != null) {
                                // 页面数据
                                val duringTime: lang.Long = pageObj.getLong("during_time")
                                val pageItem: String = pageObj.getString("item")
                                val pageItemType: String = pageObj.getString("item_type")
                                val lastPageId: String = pageObj.getString("last_page_id")
                                val pageId: String = pageObj.getString("page_id")
                                val sourceType: String = pageObj.getString("sourceType")

                                // 封装成一条页面数据
                                val pageLog: PageLog =
                                    PageLog(mid, uid, ar, ch, isNew, md, os, vc, ba, pageId, lastPageId, pageItem, pageItemType, duringTime, sourceType, ts)

                                //发送到页面访问主题
                                MyKafkaUtils.send(DWD_PAGE_TOPIC, JSON.toJSONString(pageLog, new SerializeConfig(true)))

                                //事件
                                val actionArrObj: JSONArray = jsonObj.getJSONArray("actions")
                                if (actionArrObj != null && actionArrObj.size() > 0) {
                                    for (i <- 0 until actionArrObj.size()) {
                                        val actionObj: JSONObject = actionArrObj.getJSONObject(i)
                                        //提取事件字段
                                        val actionId: String = actionObj.getString("action_id")
                                        val actionItem: String = actionObj.getString("item")
                                        val actionItemType: String = actionObj.getString("item_type")
                                        val actionTs: lang.Long = actionObj.getLong("ts")
                                        //封装成一条事件数据
                                        val pageActionLog: PageActionLog =
                                            PageActionLog(mid, uid, ar, ch, isNew, md, os, vc, ba, pageId, lastPageId, pageItem, pageItemType, duringTime, sourceType, actionId, actionItem, actionItemType, actionTs, ts)
                                        //发送事件主题
                                        MyKafkaUtils.send(DWD_PAGE_ACTION_TOPIC, JSON.toJSONString(pageActionLog, new SerializeConfig(true)))
                                    }
                                }

                                // 曝光
                            }

                            val startObj: JSONObject = jsonObj.getJSONObject("start")
                            if (startObj != null) {
                                // 启动数据
                                val entry: String = startObj.getString("entry")
                                val loadingTime: lang.Long = startObj.getLong("loading_time")
                                val openAdId: lang.Long = startObj.getLong("open_ad_id")
                                val openAdMs: lang.Long = startObj.getLong("open_ad_ms")
                                val openAdSkipMs: lang.Long = startObj.getLong("open_ad_skip_ms")

                                // 封装成一条启动数据
                                val startLog: StartLog = StartLog(mid, uid, ar, ch, isNew, md, os, vc, ba, entry, loadingTime, openAdId, openAdMs, openAdSkipMs, ts)
                                // 发送启动主题
                                MyKafkaUtils.send(DWD_START_TOPIC, JSON.toJSONString(startLog, new SerializeConfig(true)))
                            }
                        }
                    }
                )
            }
        )

        ssc.start()
        ssc.awaitTermination()

    }
}
